#!/bin/bash
#设置任务异常退出
set -eo pipefail
dt='{{ execution_date | cst_ds }}'
name='进港应集包票数和实际建包票数统计'
# 在退出时清理文件
trap "rm -f /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv && echo 'removed tmp files'" EXIT
# 创建txt文件
touch /tmp/${name}_${dt}.txt
#将中文字段名输入文件顶部
#将查询结果追加到文件中
{
  echo -e "扫描中心集散编号\t扫描中心集散名称\t包件量\t集包标识0未集包1集包\t包数量\t循环包数量">/tmp/${name}_${dt}.txt;
  spark-sql --driver-memory 8G --num-executors 20 --executor-memory 12g --executor-cores 4 --conf spark.debug.maxToStringFields=1000 --conf spark.sql.shuffle.partitions=800 --conf spark.executor.memoryOverhead=6g --conf spark.driver.maxResultSize=3g -S -e "
with send_base as (
    select waybill_no        --运单编号_ef
         , input_site        --扫描网点_ef
         , scan_site_code    --扫描网点编号_ef
         , package_code      --所属包号_ef
         , scan_time         --扫描时间_ef
         , next_station_code --下一站
    from jms_dwd.dwd_tab_barscan_send_base_hi
    where substr(dt, 1, 10) >= '{{ execution_date | cst_ds }}'
      and substr(dt, 1, 10) <= date_add('{{ execution_date | cst_ds }}', 1)
      and scan_time >= concat('{{ execution_date | cst_ds }}', ' 06:00:00')
      and scan_time <= concat(date_add('{{ execution_date | cst_ds }}', 1), ' 06:00:00')
      and substr(waybill_no, 1, 2) = 'JT'
),
     bagging_base as (
         select waybill_no                    --运单号
              , max(flight_num) as flight_num --芯片号
         from jms_dwd.dwd_tab_barscan_bagging_base_hi
         where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -9) and date_add('{{ execution_date | cst_ds }}', 1)
           and (waybill_no like 'B%' or waybill_no like '4%')
         group by waybill_no
     ),
     join_table AS (
         select t1.waybill_no                                             --运单编号_ef
              , t1.package_code                                           --所属包号_ef
              , t1.input_site                                             --扫描网点_ef
              , t1.scan_site_code                                         --扫描网点编号_ef
              , t2.network_type                     as site_net_type      --本站类型
              , t1.next_station_code                                      --下一站
              , t3.network_type                     as next_site_net_type --下一站站类型
              , t4.flight_num
              , if(t4.flight_num is not null, 1, 0) as flight_num_flag
              , if(package_code is null, 0, 1)      as package_falg
         from send_base t1
                  left join jms_dim.dim_network_whole_massage t2
                            on t1.scan_site_code = t2.code
                  left join jms_dim.dim_network_whole_massage t3
                            on t1.next_station_code = t3.code
                  left join bagging_base t4
                            on t1.package_code = t4.waybill_no
     ),
     loop_pack as (
         select scan_site_code
              , count(package_code)                   as pack_num
              , sum(if(flight_num is not null, 1, 0)) as loop_num
         from (
                  select scan_site_code --扫描网点编号
                       , package_code   --包号
                       , max(flight_num) as flight_num
                  from join_table
                  where site_net_type in (4, 5)
                    and next_site_net_type not in (4, 5)
                  group by scan_site_code --扫描网点编号
                         , package_code --包号
              ) a
         group by scan_site_code
     )
select a.scan_site_code --扫描网点编号
     , a.input_site     --扫描网点
     , a.tt1            --包件量
     , a.bnj            --是否集包 0表示未集包、1表示已集包
     , b.pack_num
     , b.loop_num
from (
         select scan_site_code                        --扫描网点编号
              , input_site                            --扫描网点
              , count(scan_site_code)          as tt1 --包件量
              , if(package_code is null, 0, 1) as bnj --是否集包 0表示未集包、1表示已集包
         from join_table
         where site_net_type in (4, 5)
           and next_site_net_type not in (4, 5)
         group by scan_site_code --扫描网点编号
                , input_site     --扫描网点
                , if(package_code is null, 0, 1)
     ) a
         left join loop_pack b
                   on a.scan_site_code = b.scan_site_code
 " >>/tmp/${name}_${dt}.txt
}

# 替换分隔符，并输出csv
cat /tmp/${name}_${dt}.txt | tr -s '\t' ',' > /tmp/${name}_${dt}.csv
# 压缩并删除 csv 文件
zip -jrm /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv
# 上传文件
/usr/local/python3/bin/python3  /usr/local/airflow/dags/utils/operators/upload_oss.py ${name}/${name}_${dt}.zip /tmp/${name}_${dt}.zip
# zip 文件将在进程退出后被自动清理

#声明文件大小
data_size=$(ls -lh /tmp/${name}_${dt}.zip | awk '{print $5}')
#打印文件信息
Message="${name}_${dt}.zip 上传成功 大小: $data_size"
#配置钉钉告警
curl 'https://oapi.dingtalk.com/robot/send?access_token=816a8f35fd84432ffa6bff9d7714abc322628bb4a8ded86908550d2335aba613' \
   -H 'Content-Type: application/json' \
   -d "
  {\"msgtype\": \"text\",
    \"text\": {
        \"content\": \"监控内容:$Message\"
     }
  }" || true
rm -f  /tmp/${name}_${dt}.txt